package cn.zglbk.network.rpc;

import cn.zglbk.network.param.GameMessageHeader;
import cn.zglbk.network.param.GameMessagePackage;
import cn.zglbk.network.param.IGameMessage;
import io.netty.util.concurrent.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.atomic.AtomicInteger;

public class GameRpcService {

    private AtomicInteger seqId = new AtomicInteger();// 自增的唯一序列Id
    private int localServerId;// 本地服务实例ID

    private EventExecutorGroup eventExecutorGroup;
    private static Logger logger = LoggerFactory.getLogger(GameRpcService.class);
    private GameRpcCallbackService gameRpcCallbackService;
    private String requestTopic;
    private String responseTopic;

    public GameRpcService(String requestTopic, String responseTopic, int localServerId, EventExecutorGroup eventExecutorGroup) {
        this.localServerId = localServerId;
        this.requestTopic = requestTopic;
        this.responseTopic = responseTopic;
        this.eventExecutorGroup = eventExecutorGroup;
        this.gameRpcCallbackService = new GameRpcCallbackService(eventExecutorGroup);
    }
    public void sendRPCResponse(IGameMessage gameMessage) {
        GameMessagePackage gameMessagePackage = new GameMessagePackage();
        gameMessagePackage.setHeader(gameMessage.getHeader());
        gameMessagePackage.setBody(gameMessage.body());
//        String sendTopic = TopicUtils.generateTopic(responseTopic, gameMessage.getHeader().getToServerId());
//        byte[] value = GameMessageInnerDecoder.sendMessage(gameMessagePackage);
//        ProducerRecord<String, byte[]> record = new ProducerRecord<String, byte[]>(sendTopic, String.valueOf(gameMessage.getHeader().getPlayerId()), value);
//        kafkaTemplate.send(record);
    }
    public void recieveResponse(IGameMessage gameMessage) {
        gameRpcCallbackService.callback(gameMessage);
    }

    public void sendRPCRequest(IGameMessage gameMessage, Promise<IGameMessage> promise) {
        GameMessagePackage gameMessagePackage = new GameMessagePackage();
        gameMessagePackage.setHeader(gameMessage.getHeader());
        gameMessagePackage.setBody(gameMessage.body());
        GameMessageHeader header = gameMessage.getHeader();
        header.setClientSeqId(seqId.incrementAndGet());
        header.setFromServerId(localServerId);
        header.setClientSendTime(System.currentTimeMillis());
        long playerId = header.getPlayerId();
        int serviceId = header.getServiceId();

//        Promise<Integer> selectServerIdPromise = new DefaultPromise<>(this.eventExecutorGroup.next());
//        playerServiceInstance.selectServerId(playerId, serviceId, selectServerIdPromise).addListener(new GenericFutureListener<Future<Integer>>() {
//            @Override
//            public void operationComplete(Future<Integer> future) throws Exception {
//                if (future.isSuccess()) {
//                    header.setToServerId(future.get());
//                    // 动态创建游戏网关监听消息的topic
//                    String sendTopic = TopicUtil.generateTopic(requestTopic, gameMessage.getHeader().getToServerId());
//                    byte[] value = GameMessageInnerDecoder.sendMessage(gameMessagePackage);
//                    ProducerRecord<String, byte[]> record = new ProducerRecord<String, byte[]>(sendTopic, String.valueOf(gameMessage.getHeader().getPlayerId()), value);
//                    kafkaTemplate.send(record);
//                    gameRpcCallbackService.addCallback(header.getClientSeqId(), promise);
//                } else {
//                    logger.error("获取目标服务实例ID出错",future.cause());
//                }
//            }
//        });
    }

}
